#include "asycipc/named_pipe.hpp"

#if defined(__linux__)

#include <errno.h>
#include <fcntl.h>
#include <poll.h>
#include <signal.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/stat.h>
#include <sys/timerfd.h>
#include <sys/types.h>
#include <unistd.h>

#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstdlib>
#include <cstring>
#include <filesystem>
#include <iomanip>
#include <iostream>
#include <cmath>
#include <limits>
#include <map>
#include <mutex>
#include <queue>
#include <random>
#include <sstream>
#include <stdexcept>
#include <string>
#include <thread>
#include <unordered_map>
#include <vector>

#include "asycipc/api.hpp"
#include "asycipc/frame.hpp"
#include "asycipc/linux_diagnostics.hpp"
#include "asycipc/server_runtime.hpp"
#include "json11.hpp"

namespace asycipc {

namespace {

using json11::Json;
namespace fs = std::filesystem;

constexpr mode_t kDirectoryMode = 0700;
constexpr mode_t kFifoMode = 0600;
constexpr std::chrono::seconds kFifoStaleAge(300);
constexpr std::chrono::seconds kHealthScanInterval(5);
constexpr std::chrono::seconds kHealthAlertInterval(30);

class PosixError : public NamedPipeError {
 public:
  PosixError(const std::string& where, int err)
      : NamedPipeError(where + " failed: " + std::string(std::strerror(err))) {}
};

void ThrowPosixError(const std::string& where) {
  throw PosixError(where, errno);
}

std::string JoinPath(const fs::path& base, const std::string& name) {
  return (base / name).string();
}

void EnsureNotSymlink(const fs::path& path) {
  struct stat st {};
  if (lstat(path.c_str(), &st) != 0) {
    if (errno == ENOENT) {
      return;
    }
    ThrowPosixError("lstat(" + path.string() + ")");
  }
  if (S_ISLNK(st.st_mode)) {
    throw NamedPipeError("symlink detected at " + path.string());
  }
}

void EnsureDirectorySecure(const fs::path& path, mode_t mode) {
  EnsureNotSymlink(path);
  if (!fs::exists(path)) {
    std::error_code ec;
    fs::create_directories(path, ec);
    if (ec) {
      throw NamedPipeError("create_directories failed for " + path.string() + ": " +
                           ec.message());
    }
  }

  struct stat st {};
  if (lstat(path.c_str(), &st) != 0) {
    ThrowPosixError("lstat(" + path.string() + ")");
  }
  if (!S_ISDIR(st.st_mode)) {
    throw NamedPipeError(path.string() + " is not a directory");
  }
  if (st.st_uid != geteuid()) {
    throw NamedPipeError("directory " + path.string() + " not owned by current user");
  }
  const mode_t current_mode = st.st_mode & 0777;
  if (current_mode != mode) {
    std::cerr << "[asycipc] adjusting permissions on directory " << path
              << " from 0" << std::oct << current_mode << " to 0" << mode << std::dec
              << std::endl;
    if (chmod(path.c_str(), mode) != 0) {
      ThrowPosixError("chmod(" + path.string() + ")");
    }
  }
}

void EnsureFifoSecure(const fs::path& path, mode_t mode) {
  EnsureNotSymlink(path);
  struct stat st {};
  if (lstat(path.c_str(), &st) != 0) {
    if (errno == ENOENT) {
      if (mkfifo(path.c_str(), mode) != 0) {
        ThrowPosixError("mkfifo(" + path.string() + ")");
      }
      return;
    }
    ThrowPosixError("lstat(" + path.string() + ")");
  }

  if (!S_ISFIFO(st.st_mode)) {
    throw NamedPipeError(path.string() + " is not a FIFO");
  }
  if (st.st_uid != geteuid()) {
    throw NamedPipeError("fifo " + path.string() + " not owned by current user");
  }
  const mode_t current_mode = st.st_mode & 0777;
  if (current_mode != mode) {
    std::cerr << "[asycipc] adjusting permissions on fifo " << path
              << " from 0" << std::oct << current_mode << " to 0" << mode << std::dec
              << std::endl;
    if (chmod(path.c_str(), mode) != 0) {
      ThrowPosixError("chmod(" + path.string() + ")");
    }
  }
}

fs::path DetermineRuntimeRoot() {
  if (const char* override_dir = std::getenv("ASYIPC_RUNTIME_DIR")) {
    if (*override_dir != '\0') {
      return fs::path(override_dir);
    }
  }

  if (geteuid() == 0) {
    return fs::path("/run/asycipc");
  }

  if (const char* xdg_dir = std::getenv("XDG_RUNTIME_DIR")) {
    if (*xdg_dir != '\0') {
      return fs::path(xdg_dir) / "asycipc";
    }
  }

  fs::path user_run = fs::path("/run/user") / std::to_string(geteuid()) / "asycipc";
  std::error_code ec;
  if (fs::exists(user_run.parent_path(), ec)) {
    return user_run;
  }

  return fs::temp_directory_path() / "asycipc";
}

fs::path ResolveBasePath(const std::string& endpoint) {
  if (endpoint.empty()) {
    throw NamedPipeError("named pipe endpoint must not be empty");
  }

  fs::path base;
  if (!endpoint.empty() && endpoint[0] == '/') {
    base = fs::path(endpoint);
  } else {
    base = DetermineRuntimeRoot() / endpoint;
  }

  EnsureDirectorySecure(base, kDirectoryMode);
  return base;
}

struct LinuxHandshakeRequest {
  std::string session_id;
  pid_t client_pid = -1;
};

std::string GenerateSessionId() {
  static std::mt19937_64 rng(std::random_device{}());
  std::uniform_int_distribution<uint64_t> dist;
  const uint64_t value = dist(rng);
  std::ostringstream oss;
  oss << std::hex << std::setw(16) << std::setfill('0') << value;
  return oss.str();
}

bool IsSessionIdValid(const std::string& id) {
  if (id.empty() || id.size() > 64) {
    return false;
  }
  for (char ch : id) {
    if ((ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') ||
        (ch >= '0' && ch <= '9') || ch == '-' || ch == '_') {
      continue;
    }
    return false;
  }
  return true;
}

LinuxHandshakeRequest ParseHandshake(const std::string& payload) {
  std::string err;
  const Json json = Json::parse(payload, err);
  if (!err.empty()) {
    throw NamedPipeError("invalid handshake message: " + err);
  }
  if (json.type() != Json::OBJECT) {
    throw NamedPipeError("handshake message must be object");
  }

  LinuxHandshakeRequest request;
  const auto& obj = json.object_items();
  auto it = obj.find("session");
  if (it == obj.end() || it->second.type() != Json::STRING) {
    throw NamedPipeError("handshake missing session id");
  }
  request.session_id = it->second.string_value();
  if (!IsSessionIdValid(request.session_id)) {
    throw NamedPipeError("invalid session identifier");
  }

  auto pid_it = obj.find("pid");
  if (pid_it != obj.end()) {
    if (pid_it->second.type() != Json::NUMBER) {
      throw NamedPipeError("handshake pid must be numeric");
    }
    const double pid_value = pid_it->second.number_value();
    if (!std::isfinite(pid_value) || pid_value < 0 ||
        pid_value > static_cast<double>(std::numeric_limits<pid_t>::max())) {
      throw NamedPipeError("handshake pid out of range");
    }
    request.client_pid = static_cast<pid_t>(pid_value);
  }
  return request;
}

class LinuxPipeHandle : public PipeHandle {
 public:
  LinuxPipeHandle(int read_fd, int write_fd, fs::path read_path, fs::path write_path,
                  bool unlink_on_close)
      : read_fd_(read_fd),
        write_fd_(write_fd),
        read_path_(std::move(read_path)),
        write_path_(std::move(write_path)),
        unlink_on_close_(unlink_on_close) {}

  ~LinuxPipeHandle() override {
    Close();
  }

  void Write(const std::vector<uint8_t>& data) override {
    if (write_fd_ < 0) {
      throw NamedPipeError("pipe closed");
    }
    const uint8_t* cursor = data.data();
    std::size_t remaining = data.size();
    while (remaining > 0) {
      ssize_t written = ::write(write_fd_, cursor, remaining);
      if (written < 0) {
        if (errno == EINTR) {
          continue;
        }
        ThrowPosixError("write");
      }
      if (written == 0) {
        throw NamedPipeError("zero-byte write detected");
      }
      remaining -= static_cast<std::size_t>(written);
      cursor += written;
    }
  }

  std::vector<uint8_t> Read() override {
    if (read_fd_ < 0) {
      throw NamedPipeError("pipe closed");
    }
    std::vector<uint8_t> buffer(4096);
    while (true) {
      ssize_t received = ::read(read_fd_, buffer.data(), buffer.size());
      if (received < 0) {
        if (errno == EINTR) {
          continue;
        }
        ThrowPosixError("read");
      }
      if (received == 0) {
        return {};
      }
      buffer.resize(static_cast<std::size_t>(received));
      return buffer;
    }
  }

  int read_fd() const noexcept { return read_fd_; }

  void Close() {
    if (read_fd_ >= 0) {
      ::close(read_fd_);
      read_fd_ = -1;
    }
    if (write_fd_ >= 0) {
      ::close(write_fd_);
      write_fd_ = -1;
    }
    if (unlink_on_close_) {
      if (!read_path_.empty()) {
        ::unlink(read_path_.c_str());
      }
      if (!write_path_.empty()) {
        ::unlink(write_path_.c_str());
      }
    }
  }

 private:
  int read_fd_ = -1;
  int write_fd_ = -1;
  fs::path read_path_;
  fs::path write_path_;
  bool unlink_on_close_ = false;
};

class LinuxPipeListener : public PipeListener {
 public:
  explicit LinuxPipeListener(fs::path base_dir)
      : base_dir_(std::move(base_dir)),
        listener_path_(JoinPath(base_dir_, "listener")) {
    EnsureDirectorySecure(base_dir_, kDirectoryMode);
    EnsureFifoSecure(listener_path_, 0600);
  }

  PipeHandlePtr Accept() override {
    while (true) {
      std::string handshake;
      if (ReadHandshake(/*blocking=*/true, &handshake)) {
        return CompleteAccept(handshake);
      }
    }
  }

  fs::path base_dir() const { return base_dir_; }

  void SignalInterrupt() const {
    const int fd = ::open(listener_path_.c_str(), O_WRONLY | O_NONBLOCK | O_CLOEXEC);
    if (fd < 0) {
      return;
    }
    static constexpr char kInterrupt[] = "__interrupt__\n";
    ::write(fd, kInterrupt, sizeof(kInterrupt) - 1);
    ::close(fd);
  }

 private:
  bool ReadHandshake(bool blocking, std::string* line) {
    const int fd = OpenListenerFd(blocking);
    if (fd < 0) {
      return false;
    }

    while (true) {
      const auto pos = handshake_buffer_.find('\n');
      if (pos != std::string::npos) {
        *line = handshake_buffer_.substr(0, pos);
        handshake_buffer_.erase(0, pos + 1);
        return true;
      }

      char chunk[256];
      const ssize_t received = ::read(fd, chunk, sizeof(chunk));
      if (received < 0) {
        if (errno == EINTR) {
          continue;
        }
        if (!blocking && (errno == EAGAIN || errno == EWOULDBLOCK)) {
          return false;
        }
        ThrowPosixError("read(listener)");
      }
      if (received == 0) {
        ReopenListenerFd();
        if (!blocking) {
          return false;
        }
        continue;
      }
      handshake_buffer_.append(chunk, static_cast<std::size_t>(received));
    }
  }

  int OpenListenerFd(bool blocking) {
    if (listener_fd_ >= 0) {
      return listener_fd_;
    }

    const int flags = O_RDONLY | O_CLOEXEC | (blocking ? 0 : O_NONBLOCK);
    listener_fd_ = ::open(listener_path_.c_str(), flags);
    if (listener_fd_ < 0) {
      if (!blocking && (errno == ENXIO || errno == EAGAIN)) {
        return -1;
      }
      ThrowPosixError("open(" + listener_path_.string() + ")");
    }

    if (keepalive_fd_ < 0) {
      keepalive_fd_ = ::open(listener_path_.c_str(), O_WRONLY | O_CLOEXEC | O_NONBLOCK);
      if (keepalive_fd_ < 0 && errno != ENXIO) {
        ThrowPosixError("open(keepalive)");
      }
    }
    return listener_fd_;
  }

  void ReopenListenerFd() {
    if (listener_fd_ >= 0) {
      ::close(listener_fd_);
      listener_fd_ = -1;
    }
    handshake_buffer_.clear();
  }

  PipeHandlePtr CompleteAccept(const std::string& handshake) {
    LinuxHandshakeRequest request = ParseHandshake(handshake);
    const fs::path in_path = JoinPath(base_dir_, "session-" + request.session_id + "-in");
    const fs::path out_path = JoinPath(base_dir_, "session-" + request.session_id + "-out");

    EnsureFifoSecure(in_path, kFifoMode);
    EnsureFifoSecure(out_path, kFifoMode);

    int read_fd = ::open(in_path.c_str(), O_RDONLY | O_CLOEXEC);
    if (read_fd < 0) {
      ThrowPosixError("open(" + in_path.string() + ")");
    }

    int write_fd = ::open(out_path.c_str(), O_WRONLY | O_CLOEXEC);
    if (write_fd < 0) {
      ::close(read_fd);
      ThrowPosixError("open(" + out_path.string() + ")");
    }

    return std::make_unique<LinuxPipeHandle>(read_fd, write_fd, in_path, out_path,
                                             /*unlink_on_close=*/false);
  }

  fs::path base_dir_;
  fs::path listener_path_;
  int listener_fd_ = -1;
  int keepalive_fd_ = -1;
  std::string handshake_buffer_;
};

int OpenWithRetry(const fs::path& path, int flags, std::chrono::steady_clock::time_point deadline,
                  std::chrono::milliseconds retry_interval) {
  while (true) {
    int fd = ::open(path.c_str(), flags | O_CLOEXEC);
    if (fd >= 0) {
      return fd;
    }
    if (errno == EINTR) {
      continue;
    }
    if (errno == ENOENT || errno == ENXIO || errno == EAGAIN) {
      if (retry_interval.count() <= 0) {
        retry_interval = std::chrono::milliseconds(50);
      }
      if (deadline != std::chrono::steady_clock::time_point::max() &&
          std::chrono::steady_clock::now() + retry_interval > deadline) {
        break;
      }
      std::this_thread::sleep_for(retry_interval);
      continue;
    }
    ThrowPosixError("open(" + path.string() + ")");
  }
  throw NamedPipeError("timeout opening " + path.string());
}

void WriteAll(int fd, const std::string& data) {
  const char* cursor = data.data();
  std::size_t remaining = data.size();
  while (remaining > 0) {
    ssize_t written = ::write(fd, cursor, remaining);
    if (written < 0) {
      if (errno == EINTR) {
        continue;
      }
      ThrowPosixError("write");
    }
    if (written == 0) {
      throw NamedPipeError("zero-byte write detected");
    }
    remaining -= static_cast<std::size_t>(written);
    cursor += written;
  }
}

}  // namespace


std::unique_ptr<PipeListener> CreateNamedPipeListener(
    const NamedPipeListenerOptions& options) {
  fs::path base_dir = ResolveBasePath(options.endpoint);
  return std::make_unique<LinuxPipeListener>(std::move(base_dir));
}

PipeHandlePtr ConnectNamedPipeEndpoint(
    const NamedPipeConnectorOptions& options) {
  fs::path base_dir = ResolveBasePath(options.endpoint);
  const fs::path listener_path = base_dir / "listener";

  EnsureDirectorySecure(base_dir, kDirectoryMode);
  EnsureFifoSecure(listener_path, kFifoMode);

  const std::string session_id = GenerateSessionId();
  const fs::path in_path = base_dir / ("session-" + session_id + "-in");
  const fs::path out_path = base_dir / ("session-" + session_id + "-out");

  struct Cleanup {
    fs::path in;
    fs::path out;
    ~Cleanup() {
      if (!in.empty()) {
        ::unlink(in.c_str());
      }
      if (!out.empty()) {
        ::unlink(out.c_str());
      }
    }
  } cleanup{in_path, out_path};

  EnsureFifoSecure(in_path, kFifoMode);
  EnsureFifoSecure(out_path, kFifoMode);

  const bool infinite_timeout = options.connect_timeout.count() <= 0;
  const auto deadline = infinite_timeout
                            ? std::chrono::steady_clock::time_point::max()
                            : std::chrono::steady_clock::now() + options.connect_timeout;

  std::map<std::string, Json> handshake_object;
  handshake_object.emplace("session", Json(session_id));
  handshake_object.emplace("pid", Json(static_cast<double>(getpid())));
  Json handshake_json(handshake_object);
  std::string handshake_message = handshake_json.dump();
  handshake_message.push_back('\n');

  int listener_fd = -1;
  while (true) {
    listener_fd = ::open(listener_path.c_str(), O_WRONLY | O_NONBLOCK | O_CLOEXEC);
    if (listener_fd >= 0) {
      break;
    }
    if (errno == EINTR) {
      continue;
    }
    if (errno == ENXIO || errno == ENOENT) {
      if (!infinite_timeout && std::chrono::steady_clock::now() >= deadline) {
        throw NamedPipeError("connect named pipe timeout");
      }
      std::this_thread::sleep_for(options.retry_interval);
      continue;
    }
    ThrowPosixError("open(" + listener_path.string() + ")");
  }

  WriteAll(listener_fd, handshake_message);
  ::close(listener_fd);

  const auto retry = options.retry_interval.count() <= 0
                         ? std::chrono::milliseconds(50)
                         : options.retry_interval;

  int write_fd = OpenWithRetry(in_path, O_WRONLY, deadline, retry);
  int read_fd = OpenWithRetry(out_path, O_RDONLY, deadline, retry);

  cleanup.in.clear();
  cleanup.out.clear();

  return std::make_unique<LinuxPipeHandle>(read_fd, write_fd, in_path, out_path,
                                           /*unlink_on_close=*/false);
}

void ServeNamedPipe(PipeServer& server, PipeListener& listener,
                    NamedPipeHandshake handshake,
                    std::chrono::milliseconds maintenance_interval,
                    NamedPipeShouldContinue should_continue,
                    std::size_t /*worker_threads*/, std::size_t /*read_buffer_size*/) {
  if (!should_continue) {
    throw NamedPipeError("should_continue callback must be provided");
  }
  if (!handshake) {
    throw NamedPipeError("handshake callback must be provided");
  }

  auto* linux_listener = dynamic_cast<LinuxPipeListener*>(&listener);
  if (!linux_listener) {
    throw NamedPipeError("ServeNamedPipe requires LinuxPipeListener");
  }

  const std::filesystem::path base_dir = linux_listener->base_dir();

  int epoll_fd = epoll_create1(EPOLL_CLOEXEC);
  if (epoll_fd < 0) {
    ThrowPosixError("epoll_create1");
  }

  int event_fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
  if (event_fd < 0) {
    const int err = errno;
    ::close(epoll_fd);
    throw PosixError("eventfd", err);
  }

  int timer_fd = timerfd_create(CLOCK_MONOTONIC, TFD_CLOEXEC | TFD_NONBLOCK);
  if (timer_fd < 0) {
    const int err = errno;
    ::close(event_fd);
    ::close(epoll_fd);
    throw PosixError("timerfd_create", err);
  }

  auto close_resources = [&]() {
    ::close(timer_fd);
    ::close(event_fd);
    ::close(epoll_fd);
  };

  auto register_fd = [&](int fd, uint32_t events) {
    epoll_event ev{};
    ev.events = events;
    ev.data.fd = fd;
    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &ev) != 0) {
      ThrowPosixError("epoll_ctl");
    }
  };

  register_fd(event_fd, EPOLLIN);
  register_fd(timer_fd, EPOLLIN);

  if (maintenance_interval.count() <= 0) {
    maintenance_interval = std::chrono::milliseconds(100);
  }
  itimerspec timer_spec{};
  timer_spec.it_interval.tv_sec =
      static_cast<time_t>(maintenance_interval.count() / 1000);
  timer_spec.it_interval.tv_nsec =
      static_cast<long>((maintenance_interval.count() % 1000) * 1000000L);
  timer_spec.it_value = timer_spec.it_interval;
  if (timer_spec.it_value.tv_sec == 0 && timer_spec.it_value.tv_nsec == 0) {
    timer_spec.it_value.tv_nsec = 1000000L;
  }
  if (timerfd_settime(timer_fd, 0, &timer_spec, nullptr) != 0) {
    const int err = errno;
    close_resources();
    throw PosixError("timerfd_settime", err);
  }

  std::mutex pending_mutex;
  std::queue<PipeHandlePtr> pending;
  std::atomic<bool> stop_accept{false};
  std::atomic<int> active_connections{0};

  TelemetryCenter& telemetry = server.telemetry();
  telemetry.SetGauge("ipc_linux_connections_active", 0.0);
  telemetry.SetGauge("ipc_linux_fifo_sessions_total", 0.0);
  telemetry.SetGauge("ipc_linux_fifo_sessions_active", 0.0);
  telemetry.SetGauge("ipc_linux_fifo_sessions_incomplete", 0.0);
  telemetry.SetGauge("ipc_linux_fifo_sessions_stale", 0.0);

  auto last_health_scan = std::chrono::steady_clock::now() - kHealthScanInterval;
  linux_diagnostics::FifoHealthStats last_health_stats;
  auto last_health_alert = std::chrono::steady_clock::time_point::min();

  auto notify_event = [&]() {
    uint64_t value = 1;
    ssize_t written = ::write(event_fd, &value, sizeof(value));
    (void)written;
  };

  std::thread accept_thread([&]() {
    while (!stop_accept.load(std::memory_order_acquire)) {
      try {
        auto handle = listener.Accept();
        telemetry.IncrementCounter("ipc_linux_accept_total");
        {
          std::lock_guard<std::mutex> lock(pending_mutex);
          pending.push(std::move(handle));
        }
        notify_event();
      } catch (const std::exception& ex) {
        if (stop_accept.load(std::memory_order_acquire)) {
          break;
        }
        telemetry.IncrementCounter("ipc_linux_accept_error_total");
        server.ReportError(ex.what());
      }
    }
  });

  struct ActiveConnection {
    std::unique_ptr<ServerConnection> connection;
  };

  std::unordered_map<int, ActiveConnection> connections;
  std::vector<epoll_event> events(64);

  auto remove_connection = [&](int fd) {
    auto it = connections.find(fd);
    if (it == connections.end()) {
      return;
    }
    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, nullptr);
    it->second.connection.reset();
    connections.erase(it);
    const int remaining = active_connections.fetch_sub(1, std::memory_order_relaxed) - 1;
    telemetry.IncrementCounter("ipc_linux_disconnect_total");
    telemetry.SetGauge("ipc_linux_connections_active",
                       static_cast<double>(std::max(remaining, 0)));
  };

  auto process_pending = [&]() {
    std::queue<PipeHandlePtr> local;
    {
      std::lock_guard<std::mutex> lock(pending_mutex);
      std::swap(local, pending);
    }

    while (!local.empty()) {
      auto handle = std::move(local.front());
      local.pop();

      SessionContext session;
      bool accepted = false;
      std::string error;
      try {
        accepted = handshake(server, *handle, session);
      } catch (const std::exception& ex) {
        error = ex.what();
      } catch (...) {
        error = "unknown handshake exception";
      }

      if (!accepted) {
        telemetry.IncrementCounter("ipc_linux_handshake_failure_total");
        if (!error.empty()) {
          server.ReportError("handshake failed: " + error);
        }
        continue;
      }

      std::unique_ptr<Channel> channel;
      try {
        channel = server.CreateChannel(std::move(session));
      } catch (const std::exception& ex) {
        telemetry.IncrementCounter("ipc_linux_handshake_failure_total");
        server.ReportError(ex.what());
        continue;
      }

      telemetry.IncrementCounter("ipc_linux_handshake_success_total");

      auto connection = std::make_unique<ServerConnection>(server, std::move(handle),
                                                           std::move(channel));
      auto* linux_handle = dynamic_cast<LinuxPipeHandle*>(&connection->handle());
      if (!linux_handle) {
        server.ReportError("incompatible pipe handle for Linux ServeNamedPipe");
        continue;
      }
      const int conn_fd = linux_handle->read_fd();

      epoll_event conn_event{};
      conn_event.events = EPOLLIN | EPOLLRDHUP | EPOLLHUP | EPOLLERR;
      conn_event.data.fd = conn_fd;
      if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, conn_fd, &conn_event) != 0) {
        server.ReportError("epoll_ctl add failed: " +
                           std::string(std::strerror(errno)));
        continue;
      }

      connections.emplace(conn_fd, ActiveConnection{std::move(connection)});
      const int count = active_connections.fetch_add(1, std::memory_order_relaxed) + 1;
      telemetry.IncrementCounter("ipc_linux_connection_open_total");
      telemetry.SetGauge("ipc_linux_connections_active", static_cast<double>(count));
    }
  };

  auto drain_event_fd = [&]() {
    uint64_t value = 0;
    ssize_t result = ::read(event_fd, &value, sizeof(value));
    (void)result;
  };

  auto drain_timer_fd = [&]() {
    uint64_t expirations = 0;
    ssize_t result = ::read(timer_fd, &expirations, sizeof(expirations));
    (void)result;
  };

  while (should_continue() && !stop_accept.load(std::memory_order_acquire)) {
    int ready = epoll_wait(epoll_fd, events.data(), static_cast<int>(events.size()), -1);
    if (ready < 0) {
      if (errno == EINTR) {
        continue;
      }
      const int err = errno;
      stop_accept.store(true, std::memory_order_release);
      linux_listener->SignalInterrupt();
      if (accept_thread.joinable()) {
        accept_thread.join();
      }
      close_resources();
      throw PosixError("epoll_wait", err);
    }

    for (int i = 0; i < ready; ++i) {
      const epoll_event& ev = events[i];
      if (ev.data.fd == event_fd) {
        drain_event_fd();
        process_pending();
        continue;
      }
      if (ev.data.fd == timer_fd) {
        drain_timer_fd();
        for (auto& entry : connections) {
          entry.second.connection->FlushMaintenance();
        }
        const auto now_scan = std::chrono::steady_clock::now();
        if (now_scan - last_health_scan >= kHealthScanInterval) {
          last_health_scan = now_scan;
          auto stats = linux_diagnostics::InspectFifoHealth(base_dir, kFifoStaleAge);
          telemetry.SetGauge("ipc_linux_fifo_sessions_total",
                             static_cast<double>(stats.total_sessions));
          telemetry.SetGauge("ipc_linux_fifo_sessions_active",
                             static_cast<double>(stats.active_sessions));
          telemetry.SetGauge("ipc_linux_fifo_sessions_incomplete",
                             static_cast<double>(stats.incomplete_sessions));
          telemetry.SetGauge("ipc_linux_fifo_sessions_stale",
                             static_cast<double>(stats.stale_sessions));

          const bool anomalous = stats.incomplete_sessions > 0 || stats.stale_sessions > 0;
          const bool changed = stats.incomplete_sessions != last_health_stats.incomplete_sessions ||
                                stats.stale_sessions != last_health_stats.stale_sessions;
          const auto now_alert = std::chrono::steady_clock::now();
          if (anomalous && (changed ||
                             now_alert - last_health_alert >= kHealthAlertInterval)) {
            std::ostringstream oss;
            oss << "linux fifo health: total=" << stats.total_sessions
                << " active=" << stats.active_sessions
                << " incomplete=" << stats.incomplete_sessions
                << " stale=" << stats.stale_sessions;
            server.ReportError(oss.str());
            last_health_alert = now_alert;
          }
          last_health_stats = stats;
        }
        continue;
      }

      auto it = connections.find(ev.data.fd);
      if (it == connections.end()) {
        continue;
      }

      bool keep = true;
      if (ev.events & (EPOLLERR | EPOLLHUP | EPOLLRDHUP)) {
        keep = false;
      } else {
        try {
          keep = it->second.connection->PumpOnce();
        } catch (const std::exception& ex) {
          server.ReportError(ex.what());
          keep = false;
        }
      }

      if (!keep) {
        remove_connection(ev.data.fd);
      }
    }
  }

  stop_accept.store(true, std::memory_order_release);
  linux_listener->SignalInterrupt();
  notify_event();
  if (accept_thread.joinable()) {
    accept_thread.join();
  }

  drain_event_fd();
  process_pending();

  for (auto it = connections.begin(); it != connections.end();) {
    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, it->first, nullptr);
    it = connections.erase(it);
  }

  telemetry.SetGauge("ipc_linux_connections_active", 0.0);

  close_resources();
}

}  // namespace asycipc

#endif  // __linux__

#if !defined(_WIN32) && !defined(__linux__) && !defined(__APPLE__)

#include <chrono>
#include <string>

namespace asycipc {

std::unique_ptr<PipeListener> CreateNamedPipeListener(
    const NamedPipeListenerOptions& options) {
  (void)options;
  throw NamedPipeError("named pipes are only available on this platform");
}

PipeHandlePtr ConnectNamedPipeEndpoint(
    const NamedPipeConnectorOptions& options) {
  (void)options;
  throw NamedPipeError("named pipes are only available on this platform");
}

void ServeNamedPipe(PipeServer&, PipeListener&, NamedPipeHandshake,
                    std::chrono::milliseconds, NamedPipeShouldContinue,
                    std::size_t, std::size_t) {
  throw NamedPipeError("named pipes are only available on this platform");
}

}  // namespace asycipc

#endif
